import { Connection, Channel, Options, Replies } from "amqplib";
import { useMq } from "./useMq";
import { QueueHandler } from "./types";
import { get, set } from '@nodejs/redis'
/**
 * 监听特定主题的消息。
 * 
 * 本函数旨在建立与消息队列的连接，并订阅指定的主题。在连接或通道发生错误时，它将尝试重新连接和重新设置监听。
 * 这种设计确保了系统的健壮性和对异常情况的处理能力。
 * 
 * @param topicName 主题名称，用于发布和订阅消息。
 * @param queueName 队列名称，指定要监听的队列。
 * @param routingKey 路由键，用于过滤消息。
 * @param handler 消息处理函数，当收到消息时被调用。
 * @param options 队列选项，用于配置队列的特性。
 * @returns 返回一个函数，调用该函数可关闭连接和通道。
 */
const map: Map<string, any> = new Map()
export async function listenFromTopic(topicName: string, queueName: string, routingKey: string, handler: QueueHandler, options: Options.AssertQueue) {
    let con: Connection | undefined = undefined;
    let channel: Channel | undefined = undefined;
    let consume: Replies.Consume | undefined = undefined;
    try {
        con = await useMq()
        channel = await con.createChannel()
        await channel.assertExchange(topicName, 'topic', { durable: true }).catch()
        await channel.assertQueue(queueName, options).catch()
        await channel.bindQueue(queueName, topicName, routingKey).catch()
        channel.prefetch(20)
        consume = await channel.consume(queueName, (msg) => {
            if (msg) handler(msg, channel!).catch(e => console.error(e))
        }, { noAck: false })
        return async () => {
            if (channel) {
                if (consume) {
                    await channel.cancel(consume.consumerTag).catch(e => console.error(e));
                }
                await channel?.close().catch(e => console.error(e))
            }
            if (con) {
                await con?.close().catch(e => console.error(e))
            }
        }
    } catch (e) {
        console.error(e)
        if (channel) {
            if (consume) {
                await channel.cancel(consume.consumerTag).catch(e => console.error(e));
            }
            await channel?.close().catch(e => console.error(e))
        }
        if (con) {
            await con?.close().catch(e => console.error(e))
        }
        throw e;
    }
}